package org.tigr.htc.server.sge;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.LineNumberReader;
import java.io.PrintWriter;
import java.text.DateFormat;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.log4j.Logger;
import org.tigr.antware.shared.command.events.CommandInterrupt;
import org.tigr.antware.shared.exceptions.InvalidArgumentException;
import org.tigr.antware.shared.util.StringUtils;
import org.tigr.htc.cmd.CmdState;
import org.tigr.htc.cmd.Command;
import org.tigr.htc.cmd.CommandConfig;
import org.tigr.htc.cmd.CommandType;
import org.tigr.htc.cmd.ICommand;
import org.tigr.htc.cmd.Param;
import org.tigr.htc.cmd.ParamType;
import org.tigr.htc.common.HTCConfig;
import org.tigr.htc.request.IRequest;
import org.tigr.htc.server.IRunner;
import org.tigr.htc.server.RunnerMonitor;

/**
 * The <b>SGERunner</b> class takes a command and creates a Sun Grid Engine job 
 * and launches the job on the grid.
 * <p>
 * The class is responsible for building the submit file based on the 
 * parameters of the command and submitting the job.  Then the runner 
 * registers itself with JobMonitor.
 */
public class SgeRunner implements IRunner {
	static Logger logger = Logger.getLogger(SgeRunner.class);
	private static DateFormat df = DateFormat.getDateTimeInstance(DateFormat.SHORT, DateFormat.SHORT);
	private static boolean remonitorFile = true;
	
	/**
	 * <code> cmd </code> the command to submit to the grid
	 */
	private Command cmd;

	/**
	 * <code> cwd </code> system working directory for this runner.
	 * This is where the runner stores logs, submit files, etc.
	 */
	private File cwd;

	/**
	 * <code> CID </code> command id
	 */
	private String CID;

	/**
	 * <code> eventReader </code> reader for the event log.
	 */
	public LineNumberReader eventReader = null;

	/**
	 * <code> eventLog </code> is the file where the grid will 
	 * write events.
	 */
	public File eventLog;

	/**
	 * <code>VERY LOW</code> constant for setPriority
	 * Sge requires priorities as negative numbers by normal users
	 * Positive priorities are only allowed by privileged users
	 */
	static final int VERY_LOW = -100;

	/**
	 * <code>LOW</code> constant for setPriority
	 */
	static final int LOW = -50;

	/**
	 * <code>LOW</code> constant for setPriority
	 */
	static final int MEDIUM = -20;

	/**
	 * <code>LOW</code> constant for setPriority
	 */
	static final int HIGH = -10;

	/**
	 * <code>LOW</code> constant for setPriority
	 */
	static final int VERY_HIGH = -1;

	private static String SGE_DIRECTIVE = HTCConfig.getProperty("sge.directive");

	private boolean arrayJob;

	// Job length variables
	private static final String FAST_JOB = "fast";

	// Checkpoint directory
	private static final String CKPT_DIR = HTCConfig.getProperty("sge.ckpt.dir");

	// SGE Checkpoint
	private static final String SGE_KILL_TRAP_SCRIPT = HTCConfig.getProperty("sge.kill.trap.script");

	// Complex String to be used when a job is recheduled due to
	// exceeding the resource limits
	private StringBuffer reschedComplexStr = new StringBuffer();

	private boolean eventReadersCreated = false;

	// Create hash for file handles of an array job's event log files
	HashMap<Integer, LineNumberReader> fileHash = null;

	// Hash that keeps track of the events already processed by the monitoring code
	private HashMap<String, String> jobEventsHash = null;
	
	// Hash that stores array task hashes
	private HashMap<Integer, HashMap<String, String>> arrayTasksEventsHash = null;
	
	boolean skipNotifyToQsub = false;

	// Flag that indicates that the Monitoring thread has encountered an 'F' finish line.
	boolean jobFinished = false;

	// Flag to indicate whether project codes are enforced. Set the value of this flag based on
	// the configuration property
	private static boolean enforceProjectCodes = false;
	static {
		String strEnforceProjectCodes = HTCConfig.getProperty("enforce.project.codes");
		if (strEnforceProjectCodes != null) {
			if (strEnforceProjectCodes.equalsIgnoreCase("true")) {
				enforceProjectCodes = true;
				logger.info("Enforce project codes specified.");
			}
		}
	}
	
	/**
	 * This constructor creates a new instance of SGE runner to launch the command. The
	 * runner creates a submit file in a directory with the command ID under the specified
	 * parent directory.
	 *  
	 * @param p_cmd a <code>Command</code> object that will executed by this runner
	 * @param p_cwd a <code>File</code> representing the parent directory
	 */
	public SgeRunner(Command p_cmd, File p_cwd) {
		logger.debug("SGE runner constructor using work dir " + p_cwd);
		cmd = p_cmd;
		cwd = new File(p_cwd, "" + cmd.getID());

		// Set the arrayJob flag
		arrayJob = (cmd.getTimes() > 0) ? true : false;

		// Set initial dir to 'cwd' if it is null
		logger.debug("Command's initial directory is " + cmd.getInitialDir());
		if ( (cmd.getInitialDir() == null) || cmd.getInitialDir().equals("/")) {
			cmd.setInitialDir(cwd.getAbsolutePath());
		}
		
		// Create the hash used to keep track of monitoring the events
		if (!arrayJob) {
			jobEventsHash = new HashMap<String, String>();
		}
		else {
			arrayTasksEventsHash = new HashMap<Integer, HashMap<String, String>>();
			for (int i = 0; i < cmd.getTimes(); i++) {
				arrayTasksEventsHash.put(new Integer(i), new HashMap<String, String>());
			}
		}
		
	}

	/**
	 * The method <code>createEventReaders</code> creates the event readers required to
	 * monitor the job. If this is an array job then the array count number of readers
	 * are created.
	 * 
	 * @throws IOException
	 */
	public void createEventReaders() throws IOException {

		// Create an event log file for every job.
		eventLog = new File(cwd, "event.log");
		if (!eventLog.exists()) eventLog.createNewFile();

		// Create array job event log files.
		if (arrayJob) {
			// Create the file hash
			fileHash = new HashMap<Integer, LineNumberReader> ();
			
			// TODO: Seems like we are not incrementing the event log name which might be  
			// problematic
			for (int i = 1; i <= cmd.getTimes(); i++) {
				LineNumberReader eventReader = new LineNumberReader(
						new FileReader(eventLog));

				fileHash.put(new Integer(i), eventReader);
			}
		} else {
			eventReader = new LineNumberReader(new FileReader(eventLog));
		}
		
		eventReadersCreated = true; // Set the flag which is used by monitoring method
	}

	/**
	 * The method <code>buildArgs</code> builds a list of arguments from the specified 
	 * command object. The command object has a list of <code>Param</code> objects which
	 * may or may not contain key, but will always have values. So build the array from
	 * these param objects.
	 * 
	 * @return a <code>String</code> array with the arguments to be passed to the command
	 * invocation. 
	 */
	private String [] buildArgs() {
		logger.debug("Building arguments array.");

		String [] args = new String [cmd.getParams().size()];
		Collection params = cmd.getParams();

		Iterator iter = params.iterator();
		int i = 0;
		
		for (; iter.hasNext(); i++) {
			Param param = (Param) iter.next();
			logger.debug("Processing parameter: " + param);

			// If this parameter has a key then specify as key value pair
			// otherwise just use the value as the argument
			if (param.getType() == ParamType.PARAM) {
				if (param.getKey() != null) {
					args[i] = param.getKey() + "=" + param.getValue();
				} else {
					args[i] = param.getValue();
				}
			}
			else {
				logger.warn(" illegal/unknown parameter type for command type htc ");
			}			
		}

		logger.debug("Arguments array: " + StringUtils.toString(args));
		return args;
	}

	/**
	 * The method <code>run</code> builds a submit file for the command fires the submit
	 * event and then submits the job to the underlying grid.
	 * 
	 */
	public void run() {
		PrintWriter pw = null;
		logger.debug("Submitting job represented by command: " + cmd);

		try {
			// setup make working directory. If one already exists then delete it 
			// and recreate the directories as needed
			if (cwd.exists()) {
				logger.warn(" CWD exists, going to delete ");
				File [] files = cwd.listFiles();
				if (files.length != 0) {
					logger.warn("Directory is not empty. Deleting files first.");
					for (int i = 0; i < files.length; i++) {
						files[i].delete();
					}
				}

				// After any files in the directory have been removed, remove the 
				// directory 
				if (!cwd.delete()) {
					logger.error("Couldn't delete working directory needed by server " + cwd);
				}
			}

			if (!cwd.mkdirs()) {
				logger.error("Couldn't create working directory needed by server " + cwd);
				cmd.fireFailureEvent(new Exception(" Couldn't create working directory needed by server " + cwd));
				return;
			}

			// Create event log file for the array job
			try {
				createEventReaders();
			} catch (Exception e) {
				// TODO: handle exception
				logger.warn("Error creating the event log files for command " +cmd.getID() + " " + e.getMessage(), e);
				return;
			}

			// create writer for submit file
			File sgeSubmit = new File(cwd, "sge_job.sh");
			
			if (logger.isDebugEnabled()) {
				logger.debug("Creating submit file: " + sgeSubmit.getAbsolutePath());
			}
			
			pw = new PrintWriter(new FileWriter(sgeSubmit), true);

			initTasks();

			// create submit file
			createSubmitFile(pw);

			// Submit the command
			CID = SgeUtils.submit(cmd.getUserName(), cmd.getInitialDir(), cwd, skipNotifyToQsub);
			if (CID == null) {
				logger.warn("Error running SGE submit. Fire failure notice!");
				cmd.fireFailureEvent(
						new Exception("Command's grid ID was zero which is invalid"));
				return;
			}

			// Set the grid job ID for the command and set and update the
			// state of the command in the database so status is reflected immediately
			cmd.setGridJobID(CID);
			cmd.setState(CmdState.WAITING);
			cmd.updateCommand();

			// EVENT: Submit
			logger.info("SgeRunner sending submit event");
			//cmd.fireSubmitEvent(cmd.getSubmitURL(), "");
			cmd.fireSubmitEvent(cmd.getSubmitURL(), eventLog.getAbsolutePath());

			// monitor log file
			logger.debug("Done submitting command with HTC ID: " + cmd.getID() + 
					" cid is " + cmd.getGridJobID() + " now just wait in monitor mode");

			RunnerMonitor.register(this);

		} catch (Exception e) {
			String message = e.getMessage();
			logger.debug("Error submitting job to the Grid " +message);

			// If submission failed, check if there is any SGE error message in 
			// sge_submit.out file
			File sgeSubmitOut = new File(cwd, "sge_submit.out");
			if (sgeSubmitOut.exists()) {
				logger.debug("Reading " + sgeSubmitOut.getName() + " file for error.");
				try {
					String line = null;
					LineNumberReader fileReader = new LineNumberReader(
							new FileReader(sgeSubmitOut));
					while ((line = fileReader.readLine()) != null) {
						message = message + "\n" + line;
					}
					fileReader.close();
				} catch (Exception lne) {
					// TODO: handle exception
				}
			}

			// Build a new exception with new error message.
			Exception newException = new Exception(message);
			logger.warn("SGE Runner error " + message, newException);

			//cmd.fireFailureEvent(new Exception("Runner:" + e.getMessage()));
			cmd.fireFailureEvent(newException);

			// Added this line to persist the 'FAILED' state to the command 
			// if any exceptions occur while submitting the jobs to the grid.
			cmd.updateCommand();

			return;
		} finally {
			// always close SGE submit file
			if (pw != null) {
				pw.close();
			}
		}
	}



	/**
	 * The method <code>initTasks</code> initializes calculates the
	 * the number of tasks this command contains.  For this runner
	 * it just initializes to times.   
	 * 
	 */
	public void initTasks() {
		int tasks = cmd.getTimes();
		cmd.setNumTasks(tasks);
		cmd.getStatus().setWaiting(tasks);
	}


	private static String parseToken(Pattern pattern, String line) {
		String token = null;
		Matcher matcher = pattern.matcher(line);
		if (matcher.matches()) {
			// As the pattern matches extract the token of interest
			token = matcher.group(1);
		}

		return token;
	}
	/**
	 * The method <code>recreateCommand</code> recreates the HTC Command object from the
	 * request ID. This method parses the sge_job.sh file to recreate the command is 
	 * typically used when reconnecting to a previous request in a local HTC Service context.
	 * 
	 * @return a <code>Command</code> object representing this request.
	 * @throws IOException 
	 */
	public static Command recreateCommand(String workingDir, long htcID) throws IOException {
		logger.debug("Attempting to recreate command for htc ID: " + htcID + 
				" with working dir '" + workingDir);
		Command cmd = new Command();
		cmd.setType(CommandType.SGE);
		cmd.setID(htcID);
		CommandConfig config = new CommandConfig();


		// Setup the regular expression patterns
		Pattern outputPattern = Pattern.compile(SGE_DIRECTIVE + " -o (.+)");
		Pattern errorPattern = Pattern.compile(SGE_DIRECTIVE + " -e (.+)");
		Pattern timesPattern = Pattern.compile(SGE_DIRECTIVE + " -t 1-(\\d+)");
		Pattern runTimePattern = Pattern.compile(SGE_DIRECTIVE + " -l h_rt=(\\w+)");
		Pattern hostnamePattern = Pattern.compile(SGE_DIRECTIVE + " -l hostname=(.+)");
		Pattern archPattern = Pattern.compile(SGE_DIRECTIVE + " -l arch=(.+)");
		Pattern memoryPattern = Pattern.compile(SGE_DIRECTIVE + " -l mem_total=(\\d+)M");
		Pattern priorityPattern = Pattern.compile(SGE_DIRECTIVE + " -p (\\w+)");
		Pattern groupPattern = Pattern.compile(SGE_DIRECTIVE + " -P (\\S+)");
		Pattern evictPattern = Pattern.compile(SGE_DIRECTIVE + " -r (\\w)");
		Pattern passthroughPattern = Pattern.compile("# passthrough given !!!");
		Pattern cmdPattern = Pattern.compile("# Command: (.+)");
		Pattern argsPattern = Pattern.compile("# Args: (.+)");
		Pattern envPattern = Pattern.compile("# ENVIRONMENT: (\\w+)");
		Pattern cwdPattern = Pattern.compile("# cd (.+)");

		// create reader for submit file     
		File sgeSubmit = new File(workingDir + "/" + htcID, "sge_job.sh");
		LineNumberReader lnr = new LineNumberReader(new FileReader(sgeSubmit));
		String line = null;
		boolean passthrough = false;

		while ((line = lnr.readLine()) != null) {

			String token = null;
			if (logger.isDebugEnabled()) {
				logger.debug("Processing line: " + line);
			}
			// If the passthrough flag is true then this line should be the passthrough 
			// string so read it and reset the flag
			if (passthrough) {
				config.setPassThrough(line.substring(SGE_DIRECTIVE.length()));
				passthrough = false;
			}

			// Try to match output pattern
			if ((token = parseToken(outputPattern, line)) != null) {
				cmd.setOutput(token);
			}
			else if ((token = parseToken(envPattern, line)) != null) {
				// If user has specified any environment, set the getenv flag
				// to true.
				logger.debug("Found environment pattern: Token is " + token);

				if (token.equals("USER")){
					logger.debug("Found User Environment Pattern: ");
					config.setGetenv(true);
				}
				else if (token.equals("DEFAULT")){
					// If user has not specified any environment, set the getenv flag
					// to false, so that the default environment is used
					logger.debug("Found Default Environment Pattern: ");
					config.setGetenv(false);
				}
			}
			else if ((token = parseToken(errorPattern, line)) != null) {
				// Try to match error pattern
				cmd.setError(token);
			}
			// Try to match error pattern
			else if ((token = parseToken(timesPattern, line)) != null) {
				int times = Integer.parseInt(token);
				cmd.setTimes(times);
			}
			// Try to match running time pattern
			else if ((token = parseToken(runTimePattern, line)) != null) {
				String [] timeParts = token.split(":");
				long runningTime = Long.parseLong(timeParts[0]) * 60 + 
				Long.parseLong(timeParts[1]);
				config.setRunningTime(runningTime);
			}
			// Try to match running hostname pattern
			else if ((token = parseToken(hostnamePattern, line)) != null) {
				config.setHosts(token);
			}
			// Try to match architecture pattern
			else if ((token = parseToken(archPattern, line)) != null) {
				String [] archs = token.split("\\|");
				String [] realArchs = new String [archs.length];

				logger.debug("Arch Array : " + StringUtils.buildArrayString(archs));

				// Get the right architecture strings from configuration file
				for (int i=0; i<archs.length; i++) {
					realArchs[i] = HTCConfig.getProperty(archs[i]);
				}

				logger.debug("Real Arch Array : " + StringUtils.buildArrayString(realArchs));
				config.setOpSys(StringUtils.buildArrayString(realArchs));
			}
			// Try to match memory pattern
			else if ((token = parseToken(memoryPattern, line)) != null) {
				config.setMemory(Integer.parseInt(token));
			}
			// Try to match priority pattern
			else if ((token = parseToken(priorityPattern, line)) != null) {
				// translation of negative SGE priorities to HTC positive 
				// priority numbers
				int pri = Integer.parseInt(token);
				if (pri == VERY_LOW) {
					config.setPriority(IRequest.VERY_LOW);
				} else if (pri == LOW) {
					config.setPriority(IRequest.LOW);
				}  else if (pri == MEDIUM) {
					config.setPriority(IRequest.MEDIUM);
				}  else if (pri == HIGH) {
					config.setPriority(IRequest.HIGH);
				}  else if (pri == VERY_HIGH) {
					config.setPriority(IRequest.VERY_HIGH);
				} 
			}
			// Try to match group pattern
			else if ((token = parseToken(groupPattern, line)) != null) {
				cmd.setGroup(token);
			}
			// Try to match evict pattern
			else if ((token = parseToken(evictPattern, line)) != null) {
				if (token.equalsIgnoreCase("y")) {
					config.setEvictable(true);
				}
				else {
					config.setEvictable(false);
				}
			}
			// If the passthrough comment matches then set the passthrough flag
			else if (passthroughPattern.matcher(line).matches()) {
				passthrough = true;
			}
			// Try to match command pattern
			else if ((token = parseToken(cmdPattern, line)) != null) {
				logger.debug("Found Cmd pattern: ");
				cmd.setCommand(token);
			}
			// Try to match command's initial directory pattern
			else if ((token = parseToken(cwdPattern, line)) != null) {
				logger.debug("Found cwd pattern: Token is " +token);
				cmd.setInitialDir(token);
			}
			// Try to match args pattern
			else if ((token = parseToken(argsPattern, line)) != null) {
				// TODO: The way this code is written if one of the arguments had a = then
				// this will be split into key value, make sure that is not a problem.
				String [] args = token.split("\t"); // Split the arguments on tabs to get individual args
				for (int i = 0; i < args.length; i++) {
					String arg = args[i];
					int equalsIndex = arg.indexOf("=");
					if (equalsIndex != -1) {
						cmd.addParam(arg.substring(0, equalsIndex), arg.substring(equalsIndex + 1));
					}
					else {
						cmd.addParam(arg);
					}
				}
				//cmd.setCommand(token);
			}
		}
		lnr.close();

		// Once the command has been recreated read the event file to determine the current
		// state of the command
		logger.debug("Recreated command: " + cmd);
		cmd.setConfig(config);
		return cmd;
	}

	/**
	 * The method <code>createSubmitFile</code> creates a submit file that can be used
	 * to execute the command on the GRID
	 * 
	 * @param p_out a <code>PrintWriter</code> representing the submit file 
	 * <i>sge.submit</i> for this command.
	 */
	private void createSubmitFile(PrintWriter p_out) throws InvalidArgumentException {
		logger.debug(" in create submit file ");

		// Check to see if this request is for an array job in which case set the array 
		// flag. This is done by checking the times specified in the command
		logger.debug("times is set to " + cmd.getTimes() + " for Command " +cmd.getID());


		p_out.println("#!/bin/bash");
		p_out.println("#");
		p_out.println("# generated SGE submit file for HTC command " + cmd.getID());
		p_out.println("# submitted for user " + cmd.getUserName());
		p_out.println("# There are four sections to this file");
		p_out.println("# 1. SGE Configuration settings");
		p_out.println("# 2. Environment settings");
		p_out.println("# 3. Running outside of SGE");
		p_out.println("# 4. Run Command");

		p_out.println("############");
		p_out.println("# Section 1: SGE Configuration");
		p_out.println("############");
		p_out.println("");
		p_out.println("# SGE Interpret this script with the Bourne shell");
		p_out.println(SGE_DIRECTIVE + " -S /bin/bash");
		p_out.println("");


		p_out.println("# NOTE: you need to 'cd' to the intial dir before submitting");
		p_out.println(SGE_DIRECTIVE + " -cwd");

		p_out.println("");

		// See if the user has specified redirection for stdin and stdout  
		// in which case set those
		String output = cmd.getOutput();
		logger.debug(" Command output directory is : " + output);

		String error = cmd.getError();
		logger.debug(" Command error directory is : " + error);

		String input = cmd.getInput();
		logger.debug(" Command error directory is : " + input);

		// If this is an array job see if the output and error specification have
		// to have their index token replaced with the specific SGE task id specified
		// as the environment variable by the N1 grid engine
		if (arrayJob) {
			if (output != null) {
				output = output.replaceAll("\\$\\(Index\\)", "\\$SGE_TASK_ID");
			}

			if (error != null) {
				error = error.replaceAll("\\$\\(Index\\)", "\\$SGE_TASK_ID");
			}		    	

			if (input != null) {
				input = input.replaceAll("\\$\\(Index\\)", "\\$SGE_TASK_ID");
			}		    	
		}

		if (output != null) {
			p_out.println("# stdout goes to ");
			p_out.println(SGE_DIRECTIVE + " -o " + output);
		}

		if (error != null) {
			p_out.println("# stderr goes to ");
			p_out.println(SGE_DIRECTIVE + " -e " + error);
		}

		if (error != null && output != null && output.equals(error)) {
			// As the output and error are same set the -j flag to indicate
			p_out.println(SGE_DIRECTIVE + " -j y");
		}
		p_out.println("");


		if (arrayJob) {
			p_out.println(SGE_DIRECTIVE + " -t 1-" + cmd.getTimes() + ":1"); // Set the array specification
		}

		p_out.println();
		p_out.println("# config variables");

		CommandConfig config = cmd.getConfig();
		if (config != null) {
			// Check to see if the user has specified any running time for this 
			// request
			String runningTime = null;
			if (config.getRunningTime() > 0) {
				// If a running time has been specified, then convert that 
				// into appropriate string
				runningTime = SgeUtils.getTimeinHHmmSS(config.getRunningTime());
			}

			// If the user has specified duration then use that to determine the 
			// complex. Presently the duration is ignored as no complexes exists 
			// for duration
			if(config.getLength() != null) {
				// Get length.
				String length = config.getLength();
				logger.debug(" Length of the jobs is " + length);
			} // End of (config.getLength() != null) 

			// Build the running time into the submit options
			if (runningTime != null) {
				p_out.println(SGE_DIRECTIVE + " -l h_rt=" + runningTime);
			}

			// If the user has specified a passthrough then set that. This is done first
			// so that if the user explicitly overrode the passthrough they will be
			// honored
			if (config.getPassThrough() != null && ! config.getPassThrough().equals("")) {
				logger.debug("PassThorugh String is specified : " +config.getPassThrough());
				p_out.println("# passthrough given !!!");
				p_out.println(SGE_DIRECTIVE + " " + config.getPassThrough());
				p_out.println("");
			}

			// if there are any hosts then set the hosts requirement
			if (config.getHosts() != null) {
				p_out.println("# hostname was given as a requirement");
				p_out.println(SGE_DIRECTIVE + " -l hostname=" + config.getHosts());
				p_out.println("");
				// Add it to the reschedule complex string 
				reschedComplexStr.append(" -l hostname=" + config.getHosts());
			}

			// If the operating system or architecture is specified set that directive
			if (config.getOpSys() != null) {
				// Make sure that the user has specified a valid set of parameters
				logger.debug(" Opsys String is " + config.getOpSys());
				String[] ops = config.getOpSys().split(",");
				logger.debug(" Number of Operating systems specified is " + ops.length);

				// As the -l option seems to not be additive if they are on different lines
				// make sure that the architecture string is comma separated
				//StringBuffer sbuf = new StringBuffer(" -l arch=");
				StringBuffer sbuf = new StringBuffer();
				for (int i = 0; i < ops.length; i++) {
					if (i > 0) {
						sbuf.append("|"); // Append the comma if this is not the first entry
					}
					logger.debug(" Opsys is " + ops[i]);
					String arch = HTCConfig.getProperty(ops[i].trim().toLowerCase());
					logger.debug(" Architecture string for " + ops[i] + " is " + arch);
					sbuf.append(arch);
				}

				if(sbuf.length() != 0) {
					p_out.println(SGE_DIRECTIVE + " -l arch=" + sbuf.toString());
					p_out.println("");
					// Add it to the reschedule complex string 
					reschedComplexStr.append(" -l arch=" + sbuf.toString());
				}
			}

			logger.debug(" Setting Memory requirements");

			// If the memory requirements are specified set them, append the 'M'
			// character as the memory requirements are specified in megabytes without the
			// M in the workflow while the sun grid specification requires the 'M'
			if (config.getMemory() != 0) {
				p_out.println(SGE_DIRECTIVE + " -l mem_total=" + config.getMemory() + "M");
				// Add it to the reschedule complex string 
				reschedComplexStr.append(" -l mem_total=" + config.getMemory() + "M");
			}

			// Set the user specified priority
			// TODO: priority should be changed to queue
			// translating positive priority numbers to SGE specific
			// priority numbers
			if (config.getPriority() > 0) {
				if (config.getPriority() == IRequest.VERY_LOW) {
					p_out.println(SGE_DIRECTIVE + " -p " + VERY_LOW);
				} else if (config.getPriority() == IRequest.LOW) {
					p_out.println(SGE_DIRECTIVE + " -p " + LOW);
				} else if (config.getPriority() == IRequest.MEDIUM) {
					p_out.println(SGE_DIRECTIVE + " -p " + MEDIUM);
				} else if (config.getPriority() == IRequest.HIGH) {
					p_out.println(SGE_DIRECTIVE + " -p " + HIGH);
				}else if (config.getPriority() == IRequest.VERY_HIGH) {
					p_out.println(SGE_DIRECTIVE + " -p " + VERY_HIGH);
				} else {
					p_out.println(SGE_DIRECTIVE + " -p " + MEDIUM);
				}
			}

			// Use the group as the account string provided enforce project codes has been
			// specified and it is not null. If not group specified and enforce project codes specified
			// then throw and exception
			if (enforceProjectCodes) {
				if (cmd.getGroup() != null) {
					p_out.println(SGE_DIRECTIVE + " -P " + cmd.getGroup());
				}
				else {
					throw new InvalidArgumentException("Enforce project codes specified." + 
							" So you must specify the 'group' in DCE Specification. Cannot proceed.");
				}
			}

			// Check to see if this job is evictable. If it is evictable 
			// build into submit string. Also, set rerunnable to 'true'
			// By requesting '-r y'
			if (config.isEvictable()) {            		
				p_out.println(SGE_DIRECTIVE + " -r y ");

			}
			else {
				p_out.println(SGE_DIRECTIVE + " -r n ");
				skipNotifyToQsub = true;
			}
		} // end of (config != null)

		// Set the name of the command to the name of the actual command without path
		String cmdName = cmd.getCommand();
		int lastSepIndex = cmdName.lastIndexOf(File.separatorChar);
		cmdName = (lastSepIndex == -1) ? cmdName : cmdName.substring(lastSepIndex + 1);
		p_out.println(SGE_DIRECTIVE + " -N " + cmdName);
		p_out.println();

		p_out.println("############");
		p_out.println("# Section 2: Environment Setup Configuration");
		p_out.println("############");

		p_out.println("# ");
		p_out.println(SGE_DIRECTIVE + " -v htc_id=" + cmd.getID() + ",request_cwd=" + cwd.getAbsolutePath());
		p_out.println("");


		if (config.isGetenv()) {
			logger.debug("User has specified that the environment be propagated. So grab ENV.");

			if ((cmd.getEnv() != null) && (cmd.getEnv().size() > 0)) {
				p_out.println("# ENVIRONMENT: USER");
		    	Map<String, String> envMap = cmd.getEnv();

		    	for (String envVar : envMap.keySet()) {
					// Do not pass the environment variable 'SGE_TASK_ID' as this causes
					// problems for jobs that were distributed by some other distributed 
					// jobs.
					if (! envVar.equals("SGE_TASK_ID")) {
						p_out.println("export " + envVar + "=\"" + envMap.get(envVar) + "\"");
					}
				}
			} else {
				// Here SGE gets the environment from the submit machine, which is 
				// one of the workers.
				p_out.println("# ENVIRONMENT: DEFAULT");
				logger.debug("Let SGE get the server environment as the default.");
				p_out.println("# copy current environment");
				p_out.println(SGE_DIRECTIVE + " -V ");
			}
			p_out.println("");
		}

		p_out.println("############");
		p_out.println("# Section 3: Uncomment to run outside of sge ");
		p_out.println("############");
		p_out.println("export htc_id=" + cmd.getID());
		p_out.println("export request_cwd=" + cwd.getAbsolutePath());
		p_out.println("export CKPT_DIR=" +CKPT_DIR);
		p_out.println("# cd " + cmd.getInitialDir());
		p_out.println("sigcounter=0");
		p_out.println("");

		p_out.println("############");
		p_out.println("# Section 4: Run Command");
		p_out.println("############");

		p_out.println("trap \"sigusr2 2\" USR2");
		p_out.println("trap \"sigusr1 1\" USR1");

		// Trap method for USR2
		p_out.println("sigusr2() {");
		p_out.println("  SIGNAL=$1");
		p_out.println("  echo \"I~~~`date`~~~Job $JOB_ID caught signal $SIGNAL.~~~x~~~x~~~x\" >> ${request_cwd}/event.log");
		p_out.println("  # Apparently, SGE notifies the script multiple times about the ");
		p_out.println("  # pending KILL signal, but only reschedules it on the 1st notification");
		p_out.println("  sigcounter=`echo $(($sigcounter + 1))`");
		p_out.println("  if [ \"$sigcounter\" -eq 1 ]; then");
		p_out.println("  	  echo \"I~~~`date`~~~${htc_id}~~~Job $JOB_ID is being interrupted~~~x~~~x\" >> ${request_cwd}/event.log");
		p_out.println("       # sleeping for 10");
		p_out.println("       sleep 10");
		p_out.println("  	  echo \"I~~~`date`~~~${htc_id}~~~Process ${user_cmd_proc_id} is interrupted.~~~x~~~x\" >> ${request_cwd}/event.log");
		p_out.println("  fi");
		p_out.println("}");

		// Trap method for USR1
		p_out.println("sigusr1() {");
		p_out.println("  SIGNAL=$1");
		p_out.println("  echo \"I~~~`date` Job $JOB_ID caught signal $SIGNAL.~~~x~~~x~~~x~~~x\" >> ${request_cwd}/event.log");
		p_out.println("  MARATHON=false");
		p_out.println("  # Apparently, SGE notifies the script multiple times about the ");
		p_out.println("  # pending KILL signal, but only reschedule it on the 1st notification");
		p_out.println("  sigcounter=`echo $(($sigcounter + 1))`");
		p_out.println("  if [ \"$sigcounter\" -eq 1 ]; then");
		p_out.println("  	echo \"I~~~`date`~~~${htc_id}~~~Job $JOB_ID is being interrupted~~~x~~~x\" >> ${request_cwd}/event.log");
		p_out.println("     user_cmd_proc_id=`cat ${request_cwd}/pid.log`");
		p_out.println("  	echo \"I~~~`date`~~~${htc_id}~~~kill -2 -$user_cmd_proc_id~~~x~~~x\" >> ${request_cwd}/event.log");
		p_out.println("     kill -2 -$user_cmd_proc_pid");
		p_out.println("     # Call trap.sh, which alters the job and reschedules it");
		p_out.println(      SGE_KILL_TRAP_SCRIPT +  " $JOB_ID $SIGNAL $CKPT_DIR \"" + reschedComplexStr + "\"");
		p_out.println("     trap_ret=$?");
		p_out.println("     echo \"I~~~`date` Job $JOB_ID altered and rescheduled with return value $trap_ret~~~x~~~x~~~x~~~x\" >> ${request_cwd}/event.log");
		p_out.println("  	echo \"I~~~`date`~~~${htc_id}~~~Process ${user_cmd_proc_id} is interrupted.~~~x~~~x\" >> ${request_cwd}/event.log");
		p_out.println("  fi");
		p_out.println("}");

		p_out.println("echo \"I~~~wrapper script starting job~~~x~~~x~~~x~~~x\" >> ${request_cwd}/event.log");

		StringBuffer sgeCmd = new StringBuffer();
		p_out.println("# Command: " + cmd.getCommand());

		// If the number of times is set to be greater than zero then call the 
		// wrapper instead of the actual command so the wrapper can replace the token
		// Also tell SGE to launch an array
		sgeCmd.append(cmd.getCommand() + " ");

		// Build the list of arguments based on the command parameters and do a token
		// replacement if this is an array job
		String [] args = buildArgs();
		for (int i = 0; i < args.length; i++) {
			if (arrayJob) {
				sgeCmd.append(" " + args[i].replaceAll("\\$\\(Index\\)", "\\$\\{SGE_TASK_ID\\}"));
			}
			else {
				sgeCmd.append(" " + args[i]);
			}
		}
		// Print the arguments as a tab separated list so these can be read when
		// recreating commands
		p_out.println("# Args: " + StringUtils.buildArrayString(args, "\t"));

		// stdin is not null add it to the command
		if (input != null) {
			sgeCmd.append(" < " + input);
		}

		// Run command in background
		p_out.println(sgeCmd + " &");

		p_out.println("# Get the pid of the above job and set as an environemnt variable");
		p_out.println("user_cmd_proc_id=$!");
		p_out.println("echo \"$user_cmd_proc_id\" > ${request_cwd}/pid.log");
		p_out.println("echo \"I~~~Job Process id is ${user_cmd_proc_id}~~~x~~~x~~~x~~~x\" >> ${request_cwd}/event.log");

		p_out.println("# Wait for the above background job to finish"); 
		p_out.println("wait %%");

		p_out.println("SGE_RET_VAL=$?");

		// Get the date in UNIX epoch seconds format, to avoid trouble with locales
		p_out.println("SDATE=$(date +%s)");
		
		if (arrayJob) {
			// If this is an array job then print the task ID to the event log
			p_out.println("if [ -e \"${CKPT_DIR}/${JOB_ID}.${SGE_TASK_ID}.resched\" ]; then");
			p_out.println("    echo \"I~~~Job Rescheduled~~~x~~~x~~~x~~~x\" >> ${request_cwd}/event.log.${SGE_TASK_ID}");
			p_out.println("    rm -f ${CKPT_DIR}/${JOB_ID}.${SGE_TASK_ID}.resched");
			p_out.println("else");
			p_out.println("     echo \"F~~~${htc_id}~~~${JOB_ID}.${SGE_TASK_ID}~~~${SDATE}~~~command finished~~~${SGE_RET_VAL}\" >> ${request_cwd}/event.log.${SGE_TASK_ID}");
			p_out.println("fi");
		} else {
			// Print the job id only
			p_out.println("if [ -e \"${CKPT_DIR}/${JOB_ID}.resched\" ]; then");
			p_out.println("    echo \"I~~~Job Rescheduled~~~x~~~x~~~x~~~x\">> ${request_cwd}/event.log");
			p_out.println("    rm -f ${CKPT_DIR}/${JOB_ID}.resched");
			p_out.println("else");
			p_out.println("    echo \"F~~~${htc_id}~~~${JOB_ID}~~~${SDATE}~~~command finished~~~${SGE_RET_VAL}\" >> ${request_cwd}/event.log");
			p_out.println("fi");
		}

		p_out.println("# end of submit file ");
	}


	/* (non-Javadoc)
	 * @see org.tigr.htc.server.IRunner#getICommand()
	 */
	public ICommand getICommand() {
		return cmd;
	}

	/* (non-Javadoc)
	 * @see org.tigr.htc.server.IRunner#stop()
	 */
	public boolean stop() throws Exception {
		logger.debug("Stopping grid job " + cmd.getGridJobID());
		SgeUtils.remove(cmd.getGridJobID());

		cmd.taskInterrupted("command interrupted by user", CommandInterrupt.USER_KILL, 0);

		if (cmd.isFinished()) {
			if (cmd.getNotifyScript() != null) {
				notify("command interrupted by user", -1);
			}
		}

		// Some Commands are not marked interrupted, when issued a stop. 
		// request on them. Set the command staet to interrupted explicitly 
		// here, if the state is not INTERRUPTED.
		if (! cmd.getState().equals(CmdState.INTERRUPTED)) {
			cmd.setState(CmdState.INTERRUPTED);
		}

		return true;
	}

	/* (non-Javadoc)
	 * @see org.tigr.htc.server.IRunner#hold()
	 */
	public boolean hold() throws Exception {
		return SgeUtils.hold(CID);
	}

	/* (non-Javadoc)
	 * @see org.tigr.htc.server.IRunner#release()
	 */
	public boolean release() throws Exception {
		return SgeUtils.release(CID);
	}


	/* (non-Javadoc)
	 * @see org.tigr.htc.server.IRunner#monitor()
	 */
	public boolean isRunnerComplete() {
		// If this method is called when the readers not created it can cause a problem
		// so throw an exception.
		if (! eventReadersCreated) {
			throw new IllegalStateException("Attempt to call monitoring when readers have not been created.");
		}

		if (logger.isDebugEnabled()) {
			logger.debug("Monitor job " + cmd.getID());
			logger.debug("File size " + eventLog.length());
		}

		if ((cmd.getState() == CmdState.FINISHED) || (cmd.getState() == CmdState.INTERRUPTED)) { 
			logger.debug("Command is finished or interrupted, so don't bother checking files. Returning true.");
			return true;
		}

		try {

			if (! arrayJob) {
				
				// To address some random problems with monitoring the file we are now going to 
				// reopen the file every time we monitor the event reader. So first close the event reader
				// and reopen it
				if (eventReader != null) {
					eventReader.close();
				}				

				if (logger.isDebugEnabled()) {
					logger.debug(" reopening event log file.");
				}
				eventReader = new LineNumberReader(new FileReader(eventLog));
				
				// monitor event log file
				boolean complete = monitorForCompletion(eventReader);
				eventReader.close();

				return complete;
			} else {

				// TODO: Modify this code to work in the mode where we reopen the even file every time
				
				// Create Event log files, if the file hash is null. This could happen
				// in re-monitoring the active jobs.
				if (fileHash == null) {
					createEventReaders();
				}

				logger.debug("Monitor array job " +cmd.getID());
				// Monitor Array job files
				boolean arrayCompletion = true;

				Iterator itr = null;
				int i = 1;
				LineNumberReader arrayEventReader = null;

				for (itr = fileHash.keySet().iterator(); itr.hasNext(); i++) {
					logger.debug("EventReader " + i + " monitoring file event log file." );
					
					// Get the line number reader and monitor it
					Integer key = (Integer) itr.next();
					arrayEventReader = (LineNumberReader) fileHash.get(key);
					
					// To address some random problems with monitoring the file we are now going to 
					// reopen the files every time we monitor the event reader. So first close the event reader
					// and reopen it
					if (arrayEventReader != null) {
						arrayEventReader.close();
					}				

					if (logger.isDebugEnabled()) {
						logger.debug(" reopening event log file.");
					}
					arrayEventReader = new LineNumberReader(new FileReader(eventLog));
					
					boolean indArrayCompletion = monitorForCompletion(arrayEventReader);
					
					logger.debug("EventReader " + i + " finished monitoring event log file. Completion is " + indArrayCompletion);
					if (indArrayCompletion) {
						logger.debug("Removing event reader from the iterator.");
						itr.remove();
						arrayEventReader.close();
					}

					// Set the success of the array job
					arrayCompletion = arrayCompletion && indArrayCompletion;
					arrayEventReader = null;
				} // end of for loop

				logger.debug("Job with id "+ cmd.getID() + " is complete " + arrayCompletion);

				return arrayCompletion;
			}
		} catch (Exception e) {
			logger.debug("exception e " + e, e);
			return false;
		}

	}


	/**
	 * The method <code>monitorForCompletion</code> method checks to see if this request
	 * is complete by monitoring the event.log file for event markers. The method then fires 
	 * the appropriate events as needed.
	 * 
	 * @param eventReader
	 * @return
	 * @throws IOException a <code></code>
	 */
	private boolean monitorForCompletion(LineNumberReader eventReader) throws IOException {
		String line;
		boolean monitoringComplete = false;

		// TODO: With all the new changes make sure that this code still works in the context
		// of array jobs
		
		
		logger.debug("Monitoring file " + eventLog.getAbsolutePath() + " at line number " +
				eventReader.getLineNumber() + ", jobFinished: " + jobFinished + ", stream is ready: " +
				eventReader.ready() + ", event log length: " + eventLog.length());
		
		HashMap <String, String> eventsHash;
		
		// While there are lines from the event log and the job is not finished
		// keep monitoring. The loop terminates when jobs is finished which is 
		// caused by the monitoring code encountering a finish, terminate or error
		// event.
		while (((line = eventReader.readLine()) != null) && (! jobFinished) ) {
			
			try {
				logger.debug("New event line.");

				// Event lines have 6 (six) fields. If we cannot
				// meet this condition then something is wrong with the event log
				// for which we throw an exception.
				String[] eventFields = line.split("~~~", 6);
				
				if (eventFields.length == 6) {
					logger.debug("Event line: " + line);
					// parse line
					char type = line.charAt(0);
					
					int taskID = 0;
					int sge_id = 0;
					
					// If this runner is for an array job then read the job ID and task ID
					// else just read job ID
					if (arrayJob) {
						if (logger.isDebugEnabled()) {
							logger.debug("Job ID field: " + eventFields[2]);
						}
						String [] idFields = eventFields[2].split("\\.");
						sge_id = Integer.parseInt(idFields[0]);
						taskID = Integer.parseInt(idFields[1]);
						
						// Set the events hash to be the task event hash
						eventsHash = arrayTasksEventsHash.get(new Integer(taskID));
					} else {
						sge_id = Integer.parseInt(eventFields[2]);
						
						// Set the event hash for the entire job
						eventsHash = jobEventsHash; 
					}

					String strDate = eventFields[3];
					logger.debug("Event date from log '" + strDate + "'");

					String eventInfo = eventFields[4];

					Date eventDate = new Date(Long.parseLong(strDate) * 1000);
					String eventString = null;
					
					logger.debug("Date resulting from parsing UNIX epoch time: " + df.format(eventDate));

					switch (type) {
						case 'S':  // start event
							// Build event string stored in has associated with this event
							// Capture the host name and start time of this start line from the event.log.
							eventString = eventFields[5] + ":" + eventDate.toString();
							
							// If the start event has not been fired yet, or if it was fired but the host name 
							// and/or start time have changed, which can happen when jobs are rescheduled, then 
							// fire start event
							if (eventsHash.containsKey("Start") && eventsHash.get("Start").equals(eventString)) {
								logger.debug("Encountered a start event that has already been fired. So ignoring ...");
							}
							else {
								// Add event string to the hash and associate with start event
								eventsHash.put("Start", eventString);
								
								logger.debug("Start event with hostname " + eventFields[5]);
								
								// Fire start event
								cmd.taskStarted(eventInfo, eventDate, eventFields[5], taskID, taskID);
							}
							break;
						case 'F':  // finish event

							// Build event string stored in has associated with this event
							int taskReturnValue = Integer.parseInt(eventFields[5].trim());
							
							// Check to see if this event has already been fired in which case 
							// ignore
							 eventString = eventDate.toString() + taskReturnValue;
							
							// If the finish event has not been fired yet, or if it was fired but the return value 
							// and or time have changed, which can happen when jobs are rescheduled, then 
							// fire finish event
							if (eventsHash.containsKey("Finish") && eventsHash.get("Finish").equals(eventString)) {
								logger.debug("Encountered a finish event that has already been fired. So ignoring ...");
							}
							else {
								// Add event string to the hash and associate with finish event								
								eventsHash.put("Finish", eventString);
							
								logger.debug("Finish event with return value " + eventFields[5]);

								// Fire finish event
								cmd.taskFinished(eventInfo, eventDate, taskReturnValue, taskID);
								if (cmd.isFinished()) {
									if (cmd.getNotifyScript() != null) {
										notify(eventInfo, taskReturnValue);
									}
								}
							}
	
							// Indicate that Monitoring thread has encountered a Finish 'F' line.
							jobFinished = true;
							logger.debug("Marked the seenFinishLine flag to true.");
	
							monitoringComplete = true;
							break;
						case 'T': // info event
							logger.debug("Encountered the finish line 'T'.");
							boolean jobOnGrid = false;
							try {
								// Try to see if the job still exists in the queue on the Grid. If it does, 
								// then do not do anything. If does not, fire an interrupted event.
								DRMAAUtils.initSession();
								jobOnGrid = DRMAAUtils.doesJobExistOnGrid("" + sge_id);
	
								DRMAAUtils.exitSession();								
							} catch (Exception e) {
								logger.warn("Could not read the status of the job on grid. Error intializing DRMAA library. " + 
										"Will stop monitoring job.", e);
							}
	
							if (jobOnGrid) {
								// If the job exists on the Grid, then do not do anything.
								logger.info("Job exists on grid. May have been rescheduled. Will continue monitoring.");
							} else {
								// Check to see if this event has already been fired in which case 
								// ignore
								 eventString = eventFields[5] + ":" + eventDate.toString();
								 
								// If the start event has not been fired yet, or if it was fired but the return value 
								// and or time have changed, which can happen when jobs are rescheduled, then 
								// fire finish event
								if (eventsHash.containsKey("Terminate") && eventsHash.get("Terminate").equals(eventString)) {
									logger.debug("Encountered a finish event that has already been fired. So ignoring ...");
								}
								else {
									eventsHash.put("Terminate", eventString);
									logger.debug("Terminate event on host " + eventFields[5]);
									
									// If the job does not exist on the Grid, fire interrupt event 
									// indicating that the job exited abruptly, for some reason.
									logger.warn("Missing finish line 'F' from the job. Marking the job as failed.");
									cmd.taskFailure(new Exception("Missing finish line 'F' from the job " + sge_id + ". Marking the job as failed."), taskID);
								}								

								monitoringComplete = true;
								jobFinished = true;
							}
							break;
	
						case 'I': // Info event
							logger.debug("Info event"); 
							break;
	
						case 'E': // Failure event
							// This event does not seem to occur. Check for deprecation
							logger.debug("Failure event");
							break;
	
						default: // Unknown event
							logger.debug("Unknown event type encountered."); 
							break;
					}
				} else {
					// We are here because the event.log file could not parsed correctly
					// So we are going to terminate the monitoring as this sometimes
					// occurs when the event.log file was corrupted. For instance even though 
					// the file on the file system is OK, the line read here has binary characters
					logger.error("Invalid event log line. The line was: " + line);
					if (! remonitorFile) {
						cmd.taskFailure(new Exception("Could not parse event log file and hence cannot read state correctly. Aborting monitoring..."), 0);
						monitoringComplete = true;
						jobFinished = true;
					} else {
					    logger.debug("First pass at monitoring encountered a problem. Trying one more time from the top at the next attempt.");
					    try {
					    	eventReader.close();
					    	eventReader = null;
					    } catch (Exception e) {
					    	logger.debug("Problem closing the filehandle for the event log reader:", e);
					    }
					    eventReader = new LineNumberReader(new FileReader(eventLog));
					    monitoringComplete = false;
						remonitorFile = false;
						break;
					}
				}
			} catch (Exception e) {
				logger.debug("Exception occurred: " + e, e);
			}
		}

		logger.debug("Completed monitoring job: " + monitoringComplete);
 	
		return monitoringComplete;
	}


	// method to run notify script and/or email user
	public void notify(String p_message, int p_returnValue) {
		logger.debug("In notify.");

		if (cmd.getNotifyScript() == null) {
			logger.warn("no script set ");
			return;
		}

		String email = cmd.getEmail();
		if (email == null) {
			email = cmd.getUserName();
		}

		File scriptFile = new File(cmd.getNotifyScript());
		String cid = "-requestID=" + cmd.getID();
		String returnval = "-returnValue=" + p_returnValue;
		String lmessage = "-message=" + p_message;
		String emailArg = "-email=" + email;
		Process p;

		if (scriptFile.exists()) {
			logger.debug("Notify script file exists");
			try {
				String tmp = scriptFile.getAbsolutePath();
				logger.debug("Notify command is " + tmp);
				logger.debug("arguments " + cid + returnval + lmessage + emailArg
						+ " -- ");
				p = Runtime.getRuntime().exec(
						new String[] { scriptFile.getAbsolutePath(), cid,
								returnval, lmessage, emailArg });
				int exitVal = p.waitFor();
				logger.debug("Process exit value: " + exitVal);
			} catch (Exception e) {
				logger.debug("Exception running notify scripts " + e.getMessage(), e);
			}
		} else {
			logger.debug("Notify script doesn't exist " + cmd.getNotifyScript());
		}
	}
}

